package bookhive.catalog;

import java.util.Hashtable;
import java.util.Iterator;
import java.util.TreeSet;

import net.jxta.endpoint.Message;
import net.jxta.peergroup.PeerGroup;
import bookhive.client.ClientPeer;
import bookhive.client.ResultListener;
import bookhive.datasource.DataSource;
import bookhive.datastructures.datasource.DataItem;
import bookhive.datastructures.datasource.HashItem;
import bookhive.service.ServicePipeBuilder;
import bookhive.utils.MessageCreator;

public class RedunderPeer extends ClientPeer implements ResultListener {

	DataSource dataSource = null;

	Hashtable<String, TreeSet<String>> sources = null;

	TreeSet<String> requestedHashes = new TreeSet<String>();

	public RedunderPeer(PeerGroup pg, DataSource ds) {
		super(pg);
		dataSource = ds;
		sources = new Hashtable<String, TreeSet<String>>();
	}

	@Override
	protected void createReponseInputPipe() {

		try {
			RedunderResponseHandler reponseHandler = new RedunderResponseHandler(
					this);
			inPipeAdv = ServicePipeBuilder.createResponsePipe(group,
					"WorldCatalog:RedundatorResponseInputPipe", reponseHandler,
					3000);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void run() {
		super.run();
		startRedunding();
	}

	private void startRedunding() {
		System.out.println("Redunder PeerID = " + group.getPeerID().toString());
		int count = 1;
		while (true) {
			HashItem record = null;
			try {
				record = dataSource.getRandomRecord();
			} catch (Exception e) {
			}// e.printStackTrace();}

			if (record != null) {

				TreeSet<String> hashes = new TreeSet<String>();
				hashes.add(record.hash);
				sendDataQuery(hashes);
				requestedHashes.add(record.hash);
				if (count % 100 == 0) {
					boolean empty = false;

					synchronized (requestedHashes) {
						empty = requestedHashes.isEmpty();
					}

					while (!empty) {
						String hash = "";
						synchronized (requestedHashes) {
							hash = requestedHashes.first();
							requestedHashes.remove(hash);
						}
						count = 1;

						Iterator<DataItem> records = dataSource.sendDataQuery(
								hash).iterator();
						while (records.hasNext()) {
							sendRedundQuery(records.next());
						}
						synchronized (requestedHashes) {
							empty = requestedHashes.isEmpty();
						}
					}
				}
				count++;
			}
			try {
				Thread.sleep(10000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}

	}

	private synchronized void sendRedundQuery(DataItem record) {
		Message m = MessageCreator.createRedundateQueryMessage(record,
				inPipeAdv);
		sendMessage(m);

	}

	public synchronized void addSource(String queryData, String catalogPidString) {
		synchronized (requestedHashes) {
			requestedHashes.remove(queryData);
		}
	}

	public void newDataRecords(TreeSet<DataItem> dataItems) {
		// TODO Auto-generated method stub

	}

	public void newHashRecords(TreeSet<HashItem> hashItems) {
		// TODO Auto-generated method stub

	}

}
